package test;

import java.util.Scanner;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import preprocess.DumpClickLog;
import trouger.dmt.DumpTableRunner;
import trouger.dmt.SourceRecordFilter;
import utils.Utils;

public class ExtractClickByTime implements Tool{

	public static String getExtractClickByTimePath(String date) {
		return Utils.rootDir + date + "/ExtractClickByTime/";
	}
	
	public static class SourceFilter implements SourceRecordFilter {
		private static final int timeSpan = 180;
		@Override
		public boolean isValidRecord(int sourceIndex, String[] segment) {
			return Utils.getTimeSpan(segment[8], "20100531200000") <= timeSpan;
		}
	}
	
	@Override 
	public Configuration getConf() {
		return null;
	}
	
	@Override
	public void setConf(Configuration conf) {
		
	}
	/*
	 * (non-Javadoc)
	 * ExtractClickByTimeed Table
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/ExtractClickByTime/
	 * 1.user 32位id
	 * 2.Query 
	 * 3.bucket号（占用Query_type，如果对应不到bucket号，则为空，其它为0-19的值）
	 * 4.User 数字id
	 * 5.用户分类
	 * 6.Machine id
	 * 7.Session ID
	 * 8.点击时间
	 * 9.搜索的来源pid
	 * 10.搜索类目
	 * 11.上一步点击模块ID（智能导航或相关搜索）
	 * 12.排序, 格式以转换
	 * 13.列表模式（大图、小图）
	 * 14.消保筛选
	 * 15.所在地筛选
	 * 16.每页商品数
	 * 17.第几页
	 * 18.被点击的商品在商品列表的第几个位置
	 * 19.点击商品的auctionId(32位ID)
	 * 20.商品所属类目，用auctionId关联
	 * 21.商品所属属性，用auctionId关联 (目前Hadoop上没有该字段，需要导入)
	 * 22.商品所属的spuid
	 * 23.商品所属卖家ID
	 * 24.B/C商品标识
	 * 25.搜索属性
	 * 26.Tab类型 (0:未知，1：所有宝贝，2：人气，3：淘宝商城，4：二手)
	 * 27.是否豆腐块商品（0:否，1：是） 
	 * 注：wiki中描述有28列是错的，实际只有27列，最后一列无效
	 */
	@Override
	public int run(String[] args) throws Exception {
		String clickLogFile = DumpClickLog.getClickLogPath(args[0]);
		String output = getExtractClickByTimePath(args[0]);
		System.out.println("start job ExtractClickByTime...");
		System.out.println("clickLogFile: " + clickLogFile);
		System.out.println("output: " + output);
		System.out.println("Continue(0/1)?");
		Scanner scanner = new Scanner(System.in);
		int c = scanner.nextInt(); 
		if (c == 0)
			return -1;
		DumpTableRunner r = new DumpTableRunner();
		r.addSourceTable(clickLogFile, "UTF-8", "\t");
		StringBuilder column = new StringBuilder();
		for (int i = 1; i < 27; ++i)
			column.append("1." + i + ", ");
		column.append("1.27");
		r.setResultFields(column.toString());
		r.setSourceRecordFilterClass(SourceFilter.class);
		r.setResultTable(output);
		r.run("ExtractClickByTimeJob");
		return 0;		 	
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		if (args.length != 1) {
			System.out.println("Usage: hadoop jar xx.jar <date>.");
			System.exit(-1);
		}
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new ExtractClickByTime(), args);
		System.exit(res);
	}
}
